package com.lagou.hbase.processor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.CollectionUtils;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;

public class RelationshipCoprocessor extends BaseRegionObserver {

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        HTableInterface htable = e.getEnvironment().getTable(TableName.valueOf("relationship"));
        byte[] row = delete.getRow();
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        for (Map.Entry<byte[], List<Cell>> entry : familyCellMap.entrySet()) {
            byte[] family = entry.getKey();
            List<Cell> cells = entry.getValue();
            if (CollectionUtils.isEmpty(cells)) {
                return;
            }
            for (Cell cell : cells) {
                byte[] qualifier = CellUtil.cloneQualifier(cell);
                //判断是否存在rowKey为delete对象的列名以及列为delete对象的rowKey,如果存在则执行删除
                Get get = new Get(qualifier);
                get.addColumn(family, row);
                if(htable.exists(get)){
                    //存在才执行删除的操作
                    Delete otherDelete = new Delete(qualifier);
                    otherDelete.addColumn(family, row);
                    htable.delete(otherDelete);
                }
            }
        }
        htable.close();
    }


}
